/*
 * Copyright 2008 Wink Saville
 * 
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License version 3
 * as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

package com.saville.mc;

import java.lang.Thread;
import java.net.*;
import java.io.*;
import java.util.HashMap;
import java.util.Map;
import java.util.LinkedList;
import java.util.concurrent.atomic.*;
import java.util.Collections;

import com.saville.debug.*;
import com.saville.msgcomp.*;
import com.saville.mc.ActiveMc;
import com.saville.mc.HtmlReq;
import com.saville.serdes.*;

public class AjaxMsgPipe {

    /*
     * Constructor
     * 
     * Open a MsgPipe using an existing socket
     */
    public AjaxMsgPipe(Socket clientSocket, int producerDstId, int producerDstSubId) {
        Log.d(df, "AjaxMsgPipe(clientSocket, producerMc) E");
        init(clientSocket, producerDstId, producerDstSubId);
        Log.d(df, "%s: AjaxMsgPipe(clientSocket, producerMc) X", mAjaxMsgPipeName);
    }

    /*
     * Constructor
     * 
     * Open a MsgPipe using the url
     */
    public AjaxMsgPipe(String clientUrl, int procuderDstId, int producerDstSubId) {
        if (!open(clientUrl, procuderDstId, producerDstSubId)) {
            throw new RuntimeException("AjaxMsgPipe: Unable to open clientUrl");
        }
    }

    /*
     * Open
     * 
     * TODO: Maybe make async and return a message when complete?
     */
    private boolean open(String clientUrl, int producerDstId, int producerDstSubId) {
        Socket clientSocket = null;
        boolean result;

        mDstId = producerDstId;
        mDstSubId = producerDstSubId;

        /* TODO: Maybe throw error rather than returning boolean? */
        try {
            URI uri = new URI(clientUrl);
            String scheme = uri.getScheme();
            int port = uri.getPort();
            if ((scheme != null) && (scheme.compareToIgnoreCase("amp") == 0)) {
                clientSocket = new Socket();
                if (port == -1) {
                    port = 46000;
                }
                Log.v(df, "%s: connect to host=%s port=%d", mAjaxMsgPipeName, uri.getHost(), port);
                clientSocket.connect(new InetSocketAddress(uri.getHost(), port));
                result = init(clientSocket, producerDstId, producerDstSubId);
            } else {
                result = false;
            }
        } catch (URISyntaxException uriE) {
            Log.print("%s: RISyntaxException e=%s", mAjaxMsgPipeName, uriE);
            result = false;
        } catch (IOException e) {
            Log.e(df, "Error opening socket");
            result = false;
        }

        if (!result) {
            sendClosed();
        }

        return result;
    }

    /*
     * Initialize
     * 
     * TODO: Maybe make async and return a message when complete?
     */
    private boolean init(Socket clientSocket, int dstId, int dstSubId) {
        boolean retVal;
        df = Log.ERROR;
        mMe = mInstanceCount.incrementAndGet();
        mAjaxMsgPipeName = "AjaxMsgPipe" + mMe;
        Log.d(df, "%s: init(clientSocket, producerMc) E ***", mAjaxMsgPipeName);
        mMcMgr = McMgr.getInstance();

        mSocket = clientSocket;
        mDstId = dstId;
        mDstSubId = dstSubId;

        try {
            mHtmlReq = new HtmlReq(mSocket);
            mProducer = new AjaxPipeProducer();
            retVal = true;
        } catch (Exception e) {
            Log.e(df, "%s: init(clientSocket, producerMc) Exception e=%s", mAjaxMsgPipeName, e);
            sendClosed();
            retVal = false;
        }

        Log.d(df, "%s: init(clientSocket, producerMc) X retVal=%s ***", 
            mAjaxMsgPipeName, retVal ? "OK" : "Err");
        return retVal;
    }

    /*
     * Send one and only one CMD_CLOSED
     */
    private void sendClosed() {
        if (mOpened.getAndSet(false)) {
            Msg pipeMsg = new Msg();
            pipeMsg.guid = McConst.MCPIPE_GUID;
            pipeMsg.cmd = McConst.MCPIPE_CMD_CLOSED;
            pipeMsg.dstId = mDstId;
            pipeMsg.dstSubId = mDstSubId;
            mMcMgr.sendMsg(pipeMsg);
        }
    }
    
    /*
     * This class reads the pipe of messages sent by the other side
     * and produces messages which are sent to the mDstId with mDstSubId.
     */
    private class AjaxPipeProducer implements Runnable {
        public AjaxPipeProducer() {
            mAjaxPipeProducerName = "AjaxPipeProducer" + mMe;
            Log.v(df, "%s: AjaxPipeProducer(); E", mAjaxPipeProducerName);

            Thread thr = new Thread(null, this, mAjaxPipeProducerName);
            thr.start();

            Log.v(df, "%s: AjaxPipeProducer(); X", mAjaxPipeProducerName);
        }
        
        private AjaxPipeConsumer getConsumer(String sessionIdStr) {
            /* Get consumer for sessionId TODO: Make subroutine */
            AjaxPipeConsumer consumer = null;
            for (int i = 0; i < 5; i++) {
                consumer = mConsumerSessionMap.get(sessionIdStr);
                if (consumer == null) {
                    Log.v(df, "waiting for consumer with sessionId %s", sessionIdStr);
                    sleep(250);
                }
            }
            if (consumer == null) {
                    throw new RuntimeException(String.format("%s: While registering no conumer with sessionId=%s\n",
                        mAjaxPipeProducerName, sessionIdStr));
            }
            
            return consumer;
        }
        
        public void run() {
            
            /*
             * We could handle either json or binary if we either
             * encode either when we open the stream or using a
             * header or even with each message?
             */
            SerdesJson serdes = new SerdesJson();
            
            try {
                Log.v(df, "%s: AjaxPipeProducer; run() E", mAjaxPipeProducerName);

                SocketAddress sa = mSocket.getRemoteSocketAddress();
                InetSocketAddress isa = (InetSocketAddress) sa;
                Log.v(df, "%s: remote inetAddress=%s, port=%s", mAjaxPipeProducerName, isa
                        .getAddress().getHostAddress(), isa.getPort());

                //mInStream = new DataInputStream(mSocket.getInputStream());

                while (true) {
                    mHtmlReq.start(); //GetTheRequestAndHeaders();
                    
                    String te = mHtmlReq.getHeader("Transfer-Encoding");
                    if (te != null)
                        throw new RuntimeException("Transfer-Encoding not support te='" + te + "'");
                        
                    String req_uri = mHtmlReq.getHeader("REQ_URI");
                    String req_method = mHtmlReq.getHeader("REQ_METHOD");
                    if ((req_uri.compareTo("/amp") == 0) && (req_method.compareTo("POST") == 0)) {
                        String cls = mHtmlReq.getHeader("content-length");
                        
                        if (cls != null) {
                            Msg msg;
                            
                            int len = mHtmlReq.readBody();
                            
                            serdes.init(mHtmlReq.getData(), 0, len);
                            msg = (Msg) serdes.readObject();
                            //Log.v(df, "%s: AjaxPipeProducer; got msg guid=0x%x cmd=0x%x id=%s sn=%s",
                            //   mAjaxPipeProducerName, msg.guid, msg.cmd, mRequestHash.get("x-amp-id"),
                            //   mRequestHash.get("x-amp-sn"));
                            
                            mMcMgr.sendMsg(msg);
                            
                            /* Ack the POST */
                            String responseHeaders[] = { 
                                    String.format("%s: %s\r\n", "x-amp-id", mHtmlReq.getHeader("x-amp-id")),
                                    String.format("%s: %s\r\n", "x-amp-sn", mHtmlReq.getHeader("x-amp-sn"))
                            }; 
                            mHtmlReq.sendResp(200, "application/text", responseHeaders);
                        } else {
                            /* TODO: Throw error? */
                            Log.v(df, "%s: AjaxPipeProducer; no message, ignoring", mAjaxPipeProducerName);
                            mHtmlReq.sendResp(200);
                        }
                    } else if ((req_uri.compareTo("/ampSessionId") == 0) && (req_method.compareTo("GET") == 0)) {
                        Long sessionId = mMcMgr.getSessionId();
                        Log.v(df, "%s: AjaxPipeProducer; ampSessionId amp-id=%d", mAjaxPipeProducerName, sessionId);
                        mHtmlReq.sendResp(200, "application/test", null, sessionId.toString().getBytes());
                        mConsumer = new AjaxPipeConsumer();
                        mConsumerSessionMap.put(sessionId.toString(), mConsumer);
                    } else if ((req_uri.compareTo("/ampRegister") == 0) && (req_method.compareTo("POST") == 0)) {
                        String name = mHtmlReq.readBodyAsString();
                        
                        String sessionIdStr = mHtmlReq.getHeader("x-amp-id");
                        if (sessionIdStr == null) {
                            throw new RuntimeException(String.format("%s: While registering %s sessionId=%s\n",
                                mAjaxPipeProducerName, name, sessionIdStr));
                        }
                        Log.v(df, "%s: AjaxPipeProducer; ampRegister name=%s sessionId=%s", 
                            mAjaxPipeProducerName, name, sessionIdStr);
                        
                        AjaxPipeConsumer consumer = getConsumer(sessionIdStr);
                        McId mcId = new McId();
                        mcId.name = name;
                        mcId.handler = consumer.mMcId.handler;
                        mcId = mMcMgr.registerMcId(mcId);
                        serdes.init();
                        mcId.serialize(serdes);
                        mHtmlReq.sendResp(200, "application/test", null, serdes.data());
                    } else if ((req_uri.compareTo("/ampLookup") == 0) && (req_method.compareTo("POST") == 0)) {
                        String name = mHtmlReq.readBodyAsString();
                        Log.v(df, "%s: AjaxPipeProducer; ampLookup '%s'", mAjaxPipeProducerName, name);
                        Integer id = mMcMgr.lookup(name);
                        mHtmlReq.sendResp(200, "application/test", null, id.toString().getBytes());
                    } else if ((req_uri.compareTo("/amp") == 0) && (req_method.compareTo("GET") == 0)) {
                        /* 
                         * Tell consumer that a message may be sent
                         * to the other side.
                         */
                        String cls = mHtmlReq.getHeader("content-length");
                        if (cls != null) {
                            int len = Integer.parseInt(cls);
                            if (len != 0) {
                                throw new RuntimeException("Unexpected content on amp GET");
                            }
                        }
                        
                        String sessionId = mHtmlReq.getHeader("x-amp-id");
                        AjaxPipeConsumer consumer = mConsumerSessionMap.get(sessionId);
                        Msg pipeMsg = new Msg();
                        pipeMsg.dstId = consumer.mMcId.id;
                        pipeMsg.guid = McConst.MCPIPE_GUID;
                        pipeMsg.cmd = McConst.MCPIPE_CMD_PRODUCER_HTTP_GET_MSG;
                        pipeMsg.obj = new HtmlReq(mHtmlReq.getDataOutStream(), (HashMap<String, String>)mHtmlReq.getHeaders().clone());
                        Log.v(df, "%s: AjaxPipeProducer; GET /amp for x-amp-id=%s x-get-amp-sn=%s",
                            mAjaxPipeProducerName, sessionId, mHtmlReq.getHeader("x-amp-sn"));
                        mMcMgr.sendMsg(pipeMsg);
                    } else {
                        /*
                         * A non-amp request, pass the request to on to our
                         * invoker at mDstId/mDstSubId
                         */
                        Msg pipeMsg = new Msg();
                        pipeMsg.guid = McConst.MCPIPE_GUID;
                        pipeMsg.cmd = McConst.MCPIPE_CMD_NON_AMP_CMD;
                        pipeMsg.dstId = mDstId;
                        pipeMsg.dstSubId = mDstSubId;
                        pipeMsg.obj = new HtmlReq(mHtmlReq.getDataOutStream(), (HashMap<String, String>)mHtmlReq.getHeaders().clone());
                        Log.v(df, "%s: AjaxPipeProducer; GET non-amp req",  mAjaxPipeProducerName);
                        mMcMgr.sendMsg(pipeMsg);
                    }
                }
            } catch (Exception e) {
                Log.e(df, "%s: AjaxPipeProducer; Exception e=%s", mAjaxPipeProducerName, e);
                Log.printStackTrace(e);
                sendClosed();
            } finally {
                try {
                    mHtmlReq.close();
                    //if (mInStream != null)
                    //    mInStream.close();
                } catch (Exception e) {
                    Log.e(df, "%s: AjaxPipeProducer; exception closing e=%s", mAjaxPipeProducerName, e);
                }

                Log.v(df, "%s: AjaxPipeProducer; run() X", mAjaxPipeProducerName);
            }
        }

        //private DataInputStream mInStream;
        private String mAjaxPipeProducerName;
    }

    /*
     * This class forwards messages to an the other side
     * by responding to GET messages to /amp received by
     * AjaxPipeProducer. When the producer receives the
     * GET's it queues a MCPIPE_CMD_PRODUCER_HTTP_GET_MSG
     * to this class. When these are received they are
     * saved in the mPendingGets Queue.
     * 
     * When any other message is received it is placed on
     * the mPendingMsgs Queue.
     * 
     * When there is messages on both then the next msg on
     * the mPendingMsgs queue is forwarded to the other side.
     */
    private class AjaxPipeConsumer extends ActiveMc {
        public AjaxPipeConsumer() {
            super("AjaxPipeConsumer" + mMe);
            df = Log.ERROR;

            Log.v(df, "%s: AjaxPipeConsumer(); E", mName);

            mSerdes = new SerdesJson();

            Log.v(df, "%s: AjaxPipeConsumer(); X", mName);
        }

        
        public void processMsg(Msg msg) {
            if (msg.is(McConst.MC_GUID, McConst.MC_CMD_STARTED)) {
                Log.v(df, "%s: AjaxPipeConsumer; got STARTED", mName);
            } else if (msg.is(McConst.MCPIPE_GUID, McConst.MCPIPE_CMD_PRODUCER_HTTP_GET_MSG)) {
                Log.v(df, "%s: AjaxPipeConsumer; GOT HTTP_GET_MSG", mName);
                mPendingGets.add(dupMsg(msg));
            } else {
                Log.v(df, "%s: AjaxPipeConsumer; GOT msg to forward guid=0x%x cmd=0x%x", mName,
                    msg.guid, msg.cmd);
                mPendingMsgs.add(dupMsg(msg));
            }
            
            /*
             * Forward msgs on mPendingMsgs if there is a mPendingGets msg.
             */
            Log.d(df, "mName %s", mName);
            Log.d(df, "mPendingMsgs=%s", mPendingMsgs);
            Log.d(df, "mPendingGets=%s", mPendingGets);
            Log.d(df, "%s: AjaxPipeConsumer pendingMsgs=%s pendingGets=%s",
                mName,
                mPendingMsgs.isEmpty() ? "emtpy" : "!empty",
                mPendingGets.isEmpty() ? "emtpy" : "!empty");
            while ( ! mPendingMsgs.isEmpty() && ! mPendingGets.isEmpty()) {
                HtmlReq gr = null;
                
                try {
                    Msg msgGet = mPendingGets.remove();
                    msg = mPendingMsgs.remove();
                    Log.v(df, "%s: AjaxPipeConsumer; forward msg guid=0x%x cmd=0x%x", mName,
                        msg.guid, msg.cmd);
                    
                    mSerdes.init();
                    msg.serialize(mSerdes);
                    byte data[] = mSerdes.data();
                    
                    String responseHeaders[] = new String[2];
                    gr = (HtmlReq)msgGet.obj;
                    try {
                        responseHeaders[0] = String.format("%s: %s\r\n", "x-amp-id", gr.getHeader("x-amp-id")); 
                        responseHeaders[1] = String.format("%s: %s\r\n", "x-amp-sn", gr.getHeader("x-amp-sn")); 
                    } catch (Exception e) {
                        responseHeaders = null;
                    }
                    
                    gr.sendResp(200, "application/text", responseHeaders, data);
                } catch (Exception e) {
                    Log.e(df, "%s: AjaxPipeConsumer exception processing e=%s", mName, e);
                    Log.printStackTrace(e);
                    sendClosed();
                } finally {
                    try {
                        if (gr != null) {
                            gr.close();
                        }
                    } catch (IOException ec) {
                        Log.e(df, "%s: AjaxPipeConsumer IOException closing ec=%s", mName, ec);
                    }
                }
            }

            Log.v(df, "%s: AjaxPipeConsumer processMsg() X", mName);
        }

        private SerdesJson mSerdes;
        private LinkedList<Msg> mPendingMsgs = new LinkedList<Msg>();
        private LinkedList<Msg> mPendingGets = new LinkedList<Msg>();
    }
    
    protected void sleep(int millisecs) {
        try {
            Thread.sleep(millisecs);
        } catch (InterruptedException e) {
        }
    }

    private static Map<String, AjaxPipeConsumer> mConsumerSessionMap = 
        Collections.synchronizedMap(new HashMap<String, AjaxPipeConsumer>());
    private AtomicBoolean mOpened = new AtomicBoolean(true);
    private int mDstId;
    private int mDstSubId;
    private Socket mSocket;
    private String mAjaxMsgPipeName;
    private AjaxPipeProducer mProducer;
    private AjaxPipeConsumer mConsumer;
    private HtmlReq mHtmlReq;
    private int df = Log.ERROR;
    private IMcMgr mMcMgr;
    private static AtomicInteger mInstanceCount = new AtomicInteger(0);
    private int mMe;
    
    /* Temporary for limit number of /amp GET responses */
    //private static int tc = 0;
}
